package cn.sheep.dolphin.etl

import cn.sheep.dolphin.common.DolphinAppComm
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/** 将bz2日志文件转换成parquet文件
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/11/28
  */
object Bz2Parquet {

	def main(args: Array[String]): Unit = {

		// 检验参数
		if (args.length != 2) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.etl.Bz2Parquet
				  |Param:
				  |	bz2InputPath	bz2日志文件的输入路径
				  | parquetOutPath	parquet文件的输出路径
				""".stripMargin)
			sys.exit(-1) // -1 非正常退出
		}

		// 接收参数(模式匹配了)
		val Array(bz2InputPath, parquetOutPath) = args

		val sc = DolphinAppComm.createSparkContext("将bz2日志文件转换成parquet文件")

		// 读取离线的bz2日志文件
		val data = sc.textFile(bz2InputPath)

		// 过滤非法数据
		val filteredRDD: RDD[Array[String]] = data.map(_.split(",", -1)).filter(_.size >= 85)

		// parquet <- DataFrame(几种创建方式) <- SQLContext <- RDD
		val sqlc = new SQLContext(sc)

		// 导入隐式转换(把钥匙给他)
		import cn.sheep.dolphin.bean.RichString._

		// RDD[Row] <- RDD[Array[String]]
		val rowRDD = filteredRDD.map(arr => Row(
			arr(0),
			arr(1).toIntPlus,
			arr(2).toIntPlus,
			arr(3).toIntPlus,
			arr(4).toIntPlus,
			arr(5),
			arr(6),
			arr(7).toIntPlus,
			arr(8).toIntPlus,
			arr(9).toDoublePlus,
			arr(10).toDoublePlus,
			arr(11),
			arr(12),
			arr(13),
			arr(14),
			arr(15),
			arr(16),
			arr(17).toIntPlus,
			arr(18),
			arr(19),
			arr(20).toIntPlus,
			arr(21).toIntPlus,
			arr(22),
			arr(23),
			arr(24),
			arr(25),
			arr(26).toIntPlus,
			arr(27),
			arr(28).toIntPlus,
			arr(29),
			arr(30).toIntPlus,
			arr(31).toIntPlus,
			arr(32).toIntPlus,
			arr(33),
			arr(34).toIntPlus,
			arr(35).toIntPlus,
			arr(36).toIntPlus,
			arr(37),
			arr(38).toIntPlus,
			arr(39).toIntPlus,
			arr(40).toDoublePlus,
			arr(41).toDoublePlus,
			arr(42).toIntPlus,
			arr(43),
			arr(44).toDoublePlus,
			arr(45).toDoublePlus,
			arr(46),
			arr(47),
			arr(48),
			arr(49),
			arr(50),
			arr(51),
			arr(52),
			arr(53),
			arr(54),
			arr(55),
			arr(56),
			arr(57).toIntPlus,
			arr(58).toDoublePlus,
			arr(59).toIntPlus,
			arr(60).toIntPlus,
			arr(61),
			arr(62),
			arr(63),
			arr(64),
			arr(65),
			arr(66),
			arr(67),
			arr(68),
			arr(69),
			arr(70),
			arr(71),
			arr(72),
			arr(73).toIntPlus,
			arr(74).toDoublePlus,
			arr(75).toDoublePlus,
			arr(76).toDoublePlus,
			arr(77).toDoublePlus,
			arr(78).toDoublePlus,
			arr(79),
			arr(80),
			arr(81),
			arr(82),
			arr(83),
			arr(84).toIntPlus
		))

		// schema: StructType <- demo
		val schema = StructType(Seq(
			StructField("sessionid", StringType),
			StructField("advertisersid", IntegerType),
			StructField("adorderid", IntegerType),
			StructField("adcreativeid", IntegerType),
			StructField("adplatformproviderid", IntegerType),
			StructField("sdkversion", StringType),
			StructField("adplatformkey", StringType),
			StructField("putinmodeltype", IntegerType),
			StructField("requestmode", IntegerType),
			StructField("adprice", DoubleType),
			StructField("adppprice", DoubleType),
			StructField("requestdate", StringType),
			StructField("ip", StringType),
			StructField("appid", StringType),
			StructField("appname", StringType),
			StructField("uuid", StringType),
			StructField("device", StringType),
			StructField("client", IntegerType),
			StructField("osversion", StringType),
			StructField("density", StringType),
			StructField("pw", IntegerType),
			StructField("ph", IntegerType),
			StructField("long", StringType),
			StructField("lat", StringType),
			StructField("provincename", StringType),
			StructField("cityname", StringType),
			StructField("ispid", IntegerType),
			StructField("ispname", StringType),
			StructField("networkmannerid", IntegerType),
			StructField("networkmannername",StringType),
			StructField("iseffective", IntegerType),
			StructField("isbilling", IntegerType),
			StructField("adspacetype", IntegerType),
			StructField("adspacetypename", StringType),
			StructField("devicetype", IntegerType),
			StructField("processnode", IntegerType),
			StructField("apptype", IntegerType),
			StructField("district", StringType),
			StructField("paymode", IntegerType),
			StructField("isbid", IntegerType),
			StructField("bidprice", DoubleType),
			StructField("winprice", DoubleType),
			StructField("iswin", IntegerType),
			StructField("cur", StringType),
			StructField("rate", DoubleType),
			StructField("cnywinprice", DoubleType),
			StructField("imei", StringType),
			StructField("mac", StringType),
			StructField("idfa", StringType),
			StructField("openudid", StringType),
			StructField("androidid", StringType),
			StructField("rtbprovince", StringType),
			StructField("rtbcity", StringType),
			StructField("rtbdistrict", StringType),
			StructField("rtbstreet", StringType),
			StructField("storeurl", StringType),
			StructField("realip", StringType),
			StructField("isqualityapp", IntegerType),
			StructField("bidfloor", DoubleType),
			StructField("aw", IntegerType),
			StructField("ah", IntegerType),
			StructField("imeimd5", StringType),
			StructField("macmd5", StringType),
			StructField("idfamd5", StringType),
			StructField("openudidmd5", StringType),
			StructField("androididmd5", StringType),
			StructField("imeisha1", StringType),
			StructField("macsha1", StringType),
			StructField("idfasha1", StringType),
			StructField("openudidsha1", StringType),
			StructField("androididsha1", StringType),
			StructField("uuidunknow", StringType),
			StructField("userid", StringType),
			StructField("iptype", IntegerType),
			StructField("initbidprice", DoubleType),
			StructField("adpayment", DoubleType),
			StructField("agentrate", DoubleType),
			StructField("lrate", DoubleType),
			StructField("adxrate", DoubleType),
			StructField("title", StringType),
			StructField("keywords", StringType),
			StructField("tagid", StringType),
			StructField("callbackdate", StringType),
			StructField("channelid", StringType),
			StructField("mediatype", IntegerType)
		))


		/**
		  * RDD[Row] <- RDD[Array[String]]
		  * schema: StructType <- demo
		  */
		val dataFrame = sqlc.createDataFrame(rowRDD, schema)

		// dataFrame -> parquet
		// parquet输出的时候默认采用的gz压缩格式
		dataFrame.write.parquet(parquetOutPath)

		sc.stop()
	}

}
